package io.github.junxworks.junx.stream.pipeline;

import io.github.junxworks.junx.stream.constants.DefaultParams;
import io.github.junxworks.junx.stream.extractor.DataExtractor;
import io.github.junxworks.junx.stream.extractor.ExtractContext;
import io.github.junxworks.junx.core.exception.FatalException;
import io.github.junxworks.junx.core.lifecycle.ThreadService;
import io.github.junxworks.junx.event.EventBus;
import io.github.junxworks.junx.event.impl.SimpleEventBus;
import lombok.Getter;
import lombok.Setter;

import java.util.List;
import java.util.Map;

/**
 * 数据清洗管道
 *
 * @author 王兴
 * @date 2024/09/10
 */
@Getter
@Setter
public class Pipeline extends ThreadService {

    /**
     * 流程名称
     */
    protected String flowName;
    /**
     * 事件总线
     */
    private EventBus eventBus = new SimpleEventBus();

    private DataExtractor dataExtractor;

    private ExtractContext extractContext;

    private List<BaseComponent> roots;

    private ExecFuture execFuture;

    private Map<String, Object> globalParams;

    private BaseComponent errorHandleComponent;

    private long waiteSeconds = 5;

    /**
     * @throws Throwable
     */
    @Override
    protected void onStart() throws Throwable {
        execFuture = new ExecFuture();
        globalParams.put(DefaultParams.PARAM_FUTURE, execFuture);
        try {
            roots.forEach(com -> com.accept(c -> {
                c.start();
                c.setFlowName(flowName);
                c.getEventChannelHandler().setGlobalParams(globalParams);
                try {
                    c.initialize();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
                this.eventBus.registerChannel(c.getEventChannel());
            }));
            errorHandleComponent.start();
            errorHandleComponent.initialize();
            errorHandleComponent.setFlowName(flowName);
            errorHandleComponent.getEventChannelHandler().setGlobalParams(globalParams);
            this.eventBus.registerChannel(errorHandleComponent.getEventChannel());
            eventBus.start();
            dataExtractor.setFlowName(flowName);
            dataExtractor.setEventBus(eventBus);
            extractContext.setGlobalParams(globalParams);
            dataExtractor.extract(extractContext);
        } catch (Exception e) {
            execFuture.countDown();
            throw new FatalException("数据管道执行失败", e);
        }
    }

    /**
     * @throws Throwable
     */
    @Override
    protected void onStop() throws Throwable {
        this.eventBus.stop();
    }

    /**
     * @throws Throwable
     */
    @Override
    protected void onRun() throws Throwable {
        Thread.sleep(waiteSeconds * 1000);
        if (dataExtractor.isDone() && isDone(roots) && execFuture.isNotDone()) {
            execFuture.countDown();
            this.stop();
        }
    }

    private boolean isDone(List<BaseComponent> components) {
        if (errorHandleComponent.isDone()) {
            for (BaseComponent r : components) {
                if (!r.isDone()) {
                    return false;
                }
                List<BaseComponent> children = r.getChildren();
                if (children != null && !children.isEmpty()) {
                    if (!isDone(children)) {
                        return false;
                    }
                }
            }
            return true;
        }
        return false;
    }
}
